﻿//******************************************************************************************************
//  DataSubscriber.cs - Gbtc
//
//  Copyright © 2010, Grid Protection Alliance.  All Rights Reserved.
//
//  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See
//  the NOTICE file distributed with this work for additional information regarding copyright ownership.
//  The GPA licenses this file to you under the Eclipse Public License -v 1.0 (the "License"); you may
//  not use this file except in compliance with the License. You may obtain a copy of the License at:
//
//      http://www.opensource.org/licenses/eclipse-1.0.php
//
//  Unless agreed to in writing, the subject software distributed under the License is distributed on an
//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the
//  License for the specific language governing permissions and limitations.
//
//  Code Modification History:
//  ----------------------------------------------------------------------------------------------------
//  2/2013 - University of Illinois, Information Trust Institute, Coordinated Science Laboratory
//       Updated code to work for the SIEGate system and architecture
//
//  08/20/2010 - J. Ritchie Carroll
//       Generated original version of source code.
//  02/07/2012 - Mehulbhai Thakkar
//       Modified SynchronizeMetadata to filter devices by original source and modified insert query
//       to populate OriginalSource value. Added to flag to optionally avoid metadata synchronization.
//
//******************************************************************************************************

using System.Net.Sockets;
using System.Security;
using GSF.Collections;
using GSF.Communication;
using GSF.Data;
using GSF.IO;
using GSF.Net.Security;
using GSF.Reflection;
using GSF.Security.Cryptography;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.IO;
using System.IO.Compression;
using System.Linq;
using System.Net.Security;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Xml;
using SIEGate.Core;
using SIEGate.Core.Configurator;
using SIEGate.Core.Legacy;
using SIEGate.Core.Measurements;
using UdpClient = GSF.Communication.UdpClient;

namespace GSF.TimeSeries.Transport
{
    /// <summary>
    /// Represents a data subscribing client that will connect to a data publisher for a data subscription.
    /// </summary>
    [Description("DataSubscriber: client that subscribes to a publishing server for a streaming data.")]
    [EditorBrowsable(EditorBrowsableState.Advanced)] // Normally defined as an input device protocol
    public class DataSubscriber : NetworkCommunicator, IDisposable
    {
        #region [ Members ]

        // Nested Types

        public class NetworkCommunicatorInfo : NetworkCommunicatorInfoBase
        {
            #region [ Members ]

            // Constants
            private readonly Dictionary<string, string> m_defaultSettings = new Dictionary<string, string>()
            {
                { "synchronizeMetadata", "true" },
                { "dataLossInterval", "10.0" }
            };

            // Fields
            private HashSet<Guid> m_subscribedMeasurements;
            private OperationalModes m_operationalModes;
            private DataSet m_dataSource;
            private bool m_useMillisecondResolution;
            private bool m_autoConnect;

            #endregion

            #region [ Constructors ]

            public NetworkCommunicatorInfo()
            {
                SetDefaults(m_defaultSettings);
            }

            #endregion

            #region [ Properties ]

            public HashSet<Guid> SubscribedMeasurements
            {
                get
                {
                    return m_subscribedMeasurements;
                }
                set
                {
                    m_subscribedMeasurements = value;
                }
            }

            /// <summary>
            /// Gets or sets flag that determines if <see cref="DataSubscriber"/> should attempt to autoconnection to <see cref="DataPublisher"/> using defined connection settings.
            /// </summary>
            public bool AutoConnect
            {
                get
                {
                    return m_autoConnect;
                }
                set
                {
                    m_autoConnect = value;
                }
            }

            public string CommandChannelConnectionString { get; set; }

            public double DataLossInterval
            {
                get;
                set;
            }

            public bool Internal { get; set; }

            /// <summary>
            /// Path to local certificate.
            /// </summary>
            public string LocalCertificate { get; set; }

            /// <summary>
            /// Gets or sets a set of flags that define ways in
            /// which the subscriber and publisher communicate.
            /// </summary>
            public OperationalModes OperationalModes
            {
                get
                {
                    return m_operationalModes;
                }
                set
                {
                    m_operationalModes = value;
                }
            }

            /// <summary>
            /// Path to remote certificate.
            /// </summary>
            public string RemoteCertificate
            {
                get;
                set;
            }

            public bool SynchronizeMetadata
            {
                get;
                set;
            }

            /// <summary>
            /// Gets or sets flag that informs publisher if base time-offsets can use millisecond resolution to conserve bandwidth.
            /// </summary>
            [Obsolete("SubscriptionInfo object defines this parameter.", false)]
            public bool UseMillisecondResolution
            {
                get
                {
                    return m_useMillisecondResolution;
                }
                set
                {
                    m_useMillisecondResolution = value;
                }
            }

            public X509ChainStatusFlags ValidChainFlags
            {
                get;
                set;
            }

            public SslPolicyErrors ValidPolicyErrors
            {
                get;
                set;
            }

            internal DataSet DataSource
            {
                get
                {
                    return m_dataSource;
                }
            }

            #endregion

            #region [ Methods ]

            /// <summary>
            /// Set the DataSource for this AdapterInfo class and derive member variables and properties relevant to this instance from that source.
            /// </summary>
            /// <param name="dataSource"></param>
            public override void ExtractConfiguration(DataSet dataSource)
            {
                base.ExtractConfiguration(dataSource);
                m_dataSource = dataSource;

                // If active measurements are defined, attempt to defined desired subscription points from there
                if (dataSource != null && dataSource.Tables.Contains("ActiveMeasurements"))
                {
                    try
                    {
                        // Filter to points associated with this subscriber that have been requested for subscription, are enabled and not owned locally
                        DataRow[] filteredRows = dataSource.Tables["ActiveMeasurements"].Select("Subscribed <> 0");
                        Guid signalID;

                        m_subscribedMeasurements = new HashSet<Guid>();

                        foreach (DataRow row in filteredRows)
                        {
                            // Parse primary measurement identifier
                            signalID = row["SignalID"].ToNonNullString(Guid.Empty.ToString()).ConvertToType<Guid>();
                            m_subscribedMeasurements.Add(signalID);
                        }
                    }
                    catch (Exception ex)
                    {
                        // Errors here may not be catastrophic, this simply limits the auto-assignment of input measurement keys desired for subscription
                        LogDispatcher.E("DataSubscriber", "configuration", "Failed to define subscribed measurements: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
                    }
                }

                // Make sure incoming stream IDs is empty
                IncomingStreamIDs = new HashSet<Guid>();
            }

            /// <summary>
            /// Set the member variables and properties for this AdapterInfo from a parsed configuration string
            /// </summary>
            /// <param name="configInfo">A Dictionary (string, string) that results from parsing a configuration string.</param>
            public override void Update(Dictionary<string, string> configInfo)
            {
                double interval;
                string setting;

                string localCertificate;
                string remoteCertificate;
                SslPolicyErrors validPolicyErrors;
                X509ChainStatusFlags validChainFlags;

                base.Update(configInfo);

                // Define data loss interval
                if (configInfo.TryGetValue("dataLossInterval", out setting) && double.TryParse(setting, out interval))
                    DataLossInterval = interval;

                // Set up TLS settings
                if (!configInfo.TryGetValue("localCertificate", out localCertificate) || !File.Exists(localCertificate))
                    throw new ArgumentException("The \"localCertificate\" setting must be defined and certificate file must exist when using TLS security mode.");

                LocalCertificate = localCertificate;

                if (!configInfo.TryGetValue("remoteCertificate", out remoteCertificate) || !File.Exists(remoteCertificate))
                    throw new ArgumentException("The \"remoteCertificate\" setting must be defined and certificate file must exist when using TLS security mode.");

                RemoteCertificate = remoteCertificate;

                if (configInfo.TryGetValue("validPolicyErrors", out setting) && Enum.TryParse(setting, out validPolicyErrors))
                    ValidPolicyErrors = validPolicyErrors;
                else
                    ValidPolicyErrors = SslPolicyErrors.None;

                if (configInfo.TryGetValue("validChainFlags", out setting) && Enum.TryParse(setting, out validChainFlags))
                    ValidChainFlags = validChainFlags;
                else
                    ValidChainFlags = X509ChainStatusFlags.NoError;

                // Check if synchronize metadata is disabled.
                SynchronizeMetadata = DefaultGet("synchronizeMetadata").ParseBoolean();

                // Check if measurements for this connection should be marked as "internal" - i.e., owned and allowed for proxy
                if (configInfo.TryGetValue("internal", out setting))
                    Internal = setting.ParseBoolean();

                // Check if we should be using the alternate binary format for communications with the publisher
                if (configInfo.TryGetValue("operationalModes", out setting))
                    OperationalModes = (OperationalModes)uint.Parse(setting);

                // Check if user wants to request that publisher use millisecond resolution to conserve bandwidth
                if (configInfo.TryGetValue("useMillisecondResolution", out setting))
                    UseMillisecondResolution = setting.ParseBoolean();

                // Define auto connect setting
                if (configInfo.TryGetValue("autoConnect", out setting))
                    AutoConnect = setting.ParseBoolean();

                // Get proper connection string - either from specified command channel
                // or from base connection string
                if (configInfo.TryGetValue("commandChannel", out setting))
                    CommandChannelConnectionString = setting;
                else
                    CommandChannelConnectionString = configInfo.JoinKeyValuePairs();
            }

            #endregion
        }

        // Constants
        private const MeasurementStateFlags DataRangeMask = MeasurementStateFlags.OverRangeError | MeasurementStateFlags.UnderRangeError | MeasurementStateFlags.AlarmHigh | MeasurementStateFlags.AlarmLow | MeasurementStateFlags.WarningHigh | MeasurementStateFlags.WarningLow;
        private const MeasurementStateFlags DataQualityMask = MeasurementStateFlags.BadData | MeasurementStateFlags.SuspectData | MeasurementStateFlags.FlatlineAlarm | MeasurementStateFlags.ComparisonAlarm | MeasurementStateFlags.ROCAlarm | MeasurementStateFlags.ReceivedAsBad | MeasurementStateFlags.CalculationError | MeasurementStateFlags.CalculationWarning | MeasurementStateFlags.ReservedQualityFlag;
        private const MeasurementStateFlags TimeQualityMask = MeasurementStateFlags.BadTime | MeasurementStateFlags.SuspectTime | MeasurementStateFlags.LateTimeAlarm | MeasurementStateFlags.FutureTimeAlarm | MeasurementStateFlags.UpSampled | MeasurementStateFlags.DownSampled | MeasurementStateFlags.ReservedTimeFlag;
        private const MeasurementStateFlags SystemIssueMask = MeasurementStateFlags.SystemError | MeasurementStateFlags.SystemWarning | MeasurementStateFlags.MeasurementError;
        private const MeasurementStateFlags CalculatedValueMask = MeasurementStateFlags.CalcuatedValue;
        private const MeasurementStateFlags DiscardedValueMask = MeasurementStateFlags.DiscardedValue;

        private const int EvenKey = 0;      // Even key/IV index
        private const int OddKey = 1;       // Odd key/IV index
        private const int KeyIndex = 0;     // Index of cipher key component in keyIV array
        private const int IVIndex = 1;      // Index of initialization vector component in keyIV array

        private const long MissingCacheWarningInterval = 20000000;

        // Events

        /// <summary>
        /// Occurs when client connection to the data publication server is established.
        /// </summary>
        public event EventHandler ConnectionEstablished;

        /// <summary>
        /// Occurs when client connection to the data publication server is terminated.
        /// </summary>
        public event EventHandler ConnectionTerminated;

        /// <summary>
        /// Occurs when client connection to the data publication server has successfully authenticated.
        /// </summary>
        public event EventHandler ConnectionAuthenticated;

        /// <summary>
        /// Occurs when client receives requested meta-data transmitted by data publication server.
        /// </summary>
        public event EventHandler<EventArgs<DataSet>> MetaDataReceived;

        /// <summary>
        /// Occurs when first measurement is transmitted by data publication server.
        /// </summary>
        public event EventHandler<EventArgs<Ticks>> DataStartTime;

        /// <summary>
        /// Indicates that processing for an input adapter (via temporal session) has completed.
        /// </summary>
        /// <remarks>
        /// This event is expected to only be raised when an input adapter has been designed to process
        /// a finite amount of data, e.g., reading a historical range of data during temporal procesing.
        /// </remarks>
        public event EventHandler<EventArgs<string>> ProcessingComplete;

        public event EventHandler<EventArgs<List<CompactMeasurement>>> NewMeasurements;

        // Fields
        private IClient m_commandChannel;
        private UdpClient m_dataChannel;
        private INotificationReceiver m_signalProxy;
        private System.Timers.Timer m_dataStreamMonitor;
        private long m_commandChannelConnectionAttempts;
        private long m_dataChannelConnectionAttempts;
        private volatile SignalIndexCache m_remoteSignalIndexCache;
        private volatile SignalIndexCache m_signalIndexCache;
        private volatile byte[][][] m_keyIVs;
        private volatile int m_cipherIndex;
        private volatile bool m_subscribed;
        private volatile int m_lastBytesReceived;
        private long m_monitoredBytesReceived;
        private long m_totalBytesReceived;
        private long m_lastMissingCacheWarning;
        private Guid m_nodeID;
        private int m_gatewayProtocolID;
        private int m_processingInterval;
        private List<ServerCommand> m_requests;
        private bool m_synchronizedSubscription;
        private Encoding m_encoding;
        private bool m_enabled;
        private bool m_disposed;
        private long m_processedMeasurements;

        private List<BufferBlockMeasurement> m_bufferBlockCache;
        private uint m_expectedBufferBlockSequenceNumber;

        private const string StrLogSource = "DataSubscriber";

        #endregion

        #region [ Constructors ]

        /// <summary>
        /// Creates a new <see cref="DataSubscriber"/>.
        /// </summary>
        public DataSubscriber(IMessageBusProxy busProxy, NetworkCommunicatorInfo info)
            : base(busProxy, info)
        {
            m_requests = new List<ServerCommand>();
            m_encoding = Encoding.Unicode;
            m_bufferBlockCache = new List<BufferBlockMeasurement>();
            Initialize();
        }

        /// <summary>
        /// Releases the unmanaged resources before the <see cref="DataSubscriber"/> object is reclaimed by <see cref="GC"/>.
        /// </summary>
        ~DataSubscriber()
        {
            Dispose(false);
        }

        #endregion

        #region [ Properties ]

        public new NetworkCommunicatorInfo Info
        {
            get
            {
                return (NetworkCommunicatorInfo)base.Info;
            }
            set
            {
                base.Info = value;
            }
        }

        /// <summary>
        /// Gets or sets flag that determines if <see cref="DataSubscriber"/> should attempt to autoconnection to <see cref="DataPublisher"/> using defined connection settings.
        /// </summary>
        public bool AutoConnect
        {
            get
            {
                return Info.AutoConnect;
            }
            set
            {
                Info.AutoConnect = value;
            }
        }

        /// <summary>
        /// Gets or sets flag that informs publisher if base time-offsets can use millisecond resolution to conserve bandwidth.
        /// </summary>
        [Obsolete("SubscriptionInfo object defines this parameter.", false)]
        public bool UseMillisecondResolution
        {
            get
            {
                return Info.UseMillisecondResolution;
            }
            set
            {
                Info.UseMillisecondResolution = value;
            }
        }

        /// <summary>
        /// Gets total data packet bytes received during this session.
        /// </summary>
        public long TotalBytesReceived
        {
            get
            {
                return m_totalBytesReceived;
            }
        }

        /// <summary>
        /// Gets or sets data loss monitoring interval, in seconds. Set to zero to disable monitoring.
        /// </summary>
        public double DataLossInterval
        {
            get
            {
                if ((object)m_dataStreamMonitor != null)
                    return m_dataStreamMonitor.Interval / 1000.0D;

                return 0.0D;
            }
            set
            {
                Info.DataLossInterval = value;
                if (value > 0.0D)
                {
                    if ((object)m_dataStreamMonitor == null)
                    {
                        // Create data stream monitoring timer
                        m_dataStreamMonitor = new System.Timers.Timer();
                        m_dataStreamMonitor.Elapsed += m_dataStreamMonitor_Elapsed;
                        m_dataStreamMonitor.AutoReset = true;
                        m_dataStreamMonitor.Enabled = false;
                    }
                    // Set user specified interval
                    m_dataStreamMonitor.Interval = value * 1000.0D;
                }
                else
                {
                    // Disable data monitor
                    if ((object)m_dataStreamMonitor != null)
                    {
                        m_dataStreamMonitor.Elapsed -= m_dataStreamMonitor_Elapsed;
                        m_dataStreamMonitor.Dispose();
                    }
                    m_dataStreamMonitor = null;
                }
            }
        }

        /// <summary>
        /// Gets or sets a set of flags that define ways in
        /// which the subscriber and publisher communicate.
        /// </summary>
        public OperationalModes OperationalModes
        {
            get
            {
                return Info.OperationalModes;
            }
            set
            {
                OperationalEncoding operationalEncoding;

                Info.OperationalModes = value;
                // TODO: Put m_encoding in AdapterInfo?
                operationalEncoding = (OperationalEncoding)(value & OperationalModes.EncodingMask);
                m_encoding = GetCharacterEncoding(operationalEncoding);
            }
        }

        /// <summary>
        /// Gets or sets the <see cref="OperationalEncoding"/> used by the subscriber and publisher.
        /// </summary>
        public OperationalEncoding OperationalEncoding
        {
            get
            {
                return (OperationalEncoding)(Info.OperationalModes & OperationalModes.EncodingMask);
            }
            set
            {
                Info.OperationalModes &= ~OperationalModes.EncodingMask;
                Info.OperationalModes |= (OperationalModes)value;
                m_encoding = GetCharacterEncoding(value);
            }
        }

        /// <summary>
        /// Gets or sets the desired processing interval, in milliseconds, for the adapter.
        /// </summary>
        /// <remarks>
        /// With the exception of the values of -1 and 0, this value specifies the desired processing interval for data, i.e.,
        /// basically a delay, or timer interval, overwhich to process data. A value of -1 means to use the default processing
        /// interval while a value of 0 means to process data as fast as possible.
        /// </remarks>
        public int ProcessingInterval
        {
            get
            {
                return m_processingInterval;
            }
            set
            {
                m_processingInterval = value;

                // Request server update the processing interval
                SendServerCommand(ServerCommand.UpdateProcessingInterval, EndianOrder.BigEndian.GetBytes(value));
            }
        }

        /// <summary>
        /// Gets the status of this <see cref="DataSubscriber"/>.
        /// </summary>
        /// <remarks>
        /// Derived classes should provide current status information about the adapter for display purposes.
        /// </remarks>
        public string Status
        {
            get
            {
                StringBuilder status = new StringBuilder();

                status.AppendFormat("         Subscription mode: Unsynchronized");
                status.AppendLine();
                status.AppendFormat("  Pending command requests: {0}", m_requests.Count);
                status.AppendLine();
                status.AppendFormat("                Subscribed: {0}", m_subscribed);
                status.AppendLine();
                status.AppendFormat("      Data packet security: {0}", m_keyIVs == null ? "unencrypted" : "encrypted");
                status.AppendLine();

                if (DataLossInterval > 0.0D)
                    status.AppendFormat("No data reconnect interval: {0} seconds", DataLossInterval.ToString("0.000"));
                else
                    status.Append("No data reconnect interval: disabled");

                status.AppendLine();

                if ((object)m_dataChannel != null)
                {
                    status.AppendLine();
                    status.AppendLine("Data Channel Status".CenterText(50));
                    status.AppendLine("-------------------".CenterText(50));
                    status.Append(m_dataChannel.Status);
                }

                if ((object)m_commandChannel != null)
                {
                    status.AppendLine();
                    status.AppendLine("Command Channel Status".CenterText(50));
                    status.AppendLine("----------------------".CenterText(50));
                    status.Append(m_commandChannel.Status);
                }

                return status.ToString();
            }
        }

        /// <summary>
        /// Gets or sets reference to <see cref="UdpClient"/> data channel, attaching and/or detaching to events as needed.
        /// </summary>
        protected UdpClient DataChannel
        {
            get
            {
                return m_dataChannel;
            }
            set
            {
                if (m_dataChannel != null)
                {
                    // Detach from events on existing data channel reference
                    m_dataChannel.ConnectionException -= m_dataChannel_ConnectionException;
                    m_dataChannel.ConnectionAttempt -= m_dataChannel_ConnectionAttempt;
                    m_dataChannel.ReceiveData -= m_dataChannel_ReceiveData;
                    m_dataChannel.ReceiveDataException -= m_dataChannel_ReceiveDataException;

                    if (m_dataChannel != value)
                        m_dataChannel.Dispose();
                }

                // Assign new data channel reference
                m_dataChannel = value;

                if (m_dataChannel != null)
                {
                    // Attach to desired events on new data channel reference
                    m_dataChannel.ConnectionException += m_dataChannel_ConnectionException;
                    m_dataChannel.ConnectionAttempt += m_dataChannel_ConnectionAttempt;
                    m_dataChannel.ReceiveData += m_dataChannel_ReceiveData;
                    m_dataChannel.ReceiveDataException += m_dataChannel_ReceiveDataException;
                }
            }
        }

        /// <summary>
        /// Gets or sets reference to <see cref="Communication.TcpClient"/> command channel, attaching and/or detaching to events as needed.
        /// </summary>
        protected IClient CommandChannel
        {
            get
            {
                return m_commandChannel;
            }
            set
            {
                if (m_commandChannel != null)
                {
                    // Detach from events on existing command channel reference
                    m_commandChannel.ConnectionAttempt -= m_commandChannel_ConnectionAttempt;
                    m_commandChannel.ConnectionEstablished -= m_commandChannel_ConnectionEstablished;
                    m_commandChannel.ConnectionException -= m_commandChannel_ConnectionException;
                    m_commandChannel.ConnectionTerminated -= m_commandChannel_ConnectionTerminated;
                    m_commandChannel.ReceiveData -= m_commandChannel_ReceiveData;
                    m_commandChannel.ReceiveDataException -= m_commandChannel_ReceiveDataException;
                    m_commandChannel.SendDataException -= m_commandChannel_SendDataException;

                    if (m_commandChannel != value)
                        m_commandChannel.Dispose();
                }

                // Assign new command channel reference
                m_commandChannel = value;

                if (m_commandChannel != null)
                {
                    // Attach to desired events on new command channel reference
                    m_commandChannel.ConnectionAttempt += m_commandChannel_ConnectionAttempt;
                    m_commandChannel.ConnectionEstablished += m_commandChannel_ConnectionEstablished;
                    m_commandChannel.ConnectionException += m_commandChannel_ConnectionException;
                    m_commandChannel.ConnectionTerminated += m_commandChannel_ConnectionTerminated;
                    m_commandChannel.ReceiveData += m_commandChannel_ReceiveData;
                    m_commandChannel.ReceiveDataException += m_commandChannel_ReceiveDataException;
                    m_commandChannel.SendDataException += m_commandChannel_SendDataException;
                }
            }
        }

        public bool IsConnected
        {
            get
            {
                return ((object)m_commandChannel != null && m_commandChannel.CurrentState == ClientState.Connected);
            }
        }

        public long ProcessedMeasurements
        {
            get
            {
                return m_processedMeasurements;
            }
            set
            {
                m_processedMeasurements = value;
            }
        }

        #endregion

        #region [ Methods ]

        /// <summary>
        /// Releases all the resources used by the <see cref="DataSubscriber"/> object.
        /// </summary>
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// Releases the unmanaged resources used by the <see cref="DataSubscriber"/> object and optionally releases the managed resources.
        /// </summary>
        /// <param name="disposing">true to release both managed and unmanaged resources; false to release only unmanaged resources.</param>
        protected virtual void Dispose(bool disposing)
        {
            if (!m_disposed)
            {
                try
                {
                    if (disposing)
                    {
                        DataLossInterval = 0.0D;
                        CommandChannel = null;
                        DataChannel = null;
                    }
                }
                finally
                {
                    m_disposed = true;  // Prevent duplicate dispose.
                }
            }
        }

        /// <summary>
        /// Initializes <see cref="DataSubscriber"/>.
        /// </summary>
        public void Initialize()
        {
            // Have to explicitly pull this due to side effects in DataSubscriber.DataLossInterval
            DataLossInterval = Info.DataLossInterval;

            if (Info.AutoConnect)
            {
                // Connect to local events when automatically engaging connection cycle
                ConnectionAuthenticated += DataSubscriber_ConnectionAuthenticated;
                MetaDataReceived += DataSubscriber_MetaDataReceived;
            }

            // Create a new TLS client and certificate checker
            TlsClient commandChannel = new TlsClient();
            SimpleCertificateChecker certificateChecker = new SimpleCertificateChecker();

            // Set up certificate checker
            certificateChecker.TrustedCertificates.Add(new X509Certificate2(Info.RemoteCertificate));
            certificateChecker.ValidPolicyErrors = Info.ValidPolicyErrors;
            certificateChecker.ValidChainFlags = Info.ValidChainFlags;

            // Initialize default settings
            commandChannel.PayloadAware = true;
            commandChannel.PersistSettings = false;
            commandChannel.MaxConnectionAttempts = 1;
            commandChannel.CertificateFile = Info.LocalCertificate;
            commandChannel.CertificateChecker = certificateChecker;

            // Assign command channel client reference and attach to needed events
            this.CommandChannel = commandChannel;

            m_commandChannel.ConnectionString = Info.CommandChannelConnectionString;
        }

        /// <summary>
        /// Authenticates subcriber to a data publisher.
        /// </summary>
        /// <param name="sharedSecret">Shared secret used to look up private crypto key and initialization vector.</param>
        /// <param name="authenticationID">Authentication ID that publisher will use to validate subscriber identity.</param>
        /// <returns><c>true</c> if authentication transmission was successful; otherwise <c>false</c>.</returns>
        public virtual bool Authenticate(string sharedSecret, string authenticationID)
        {
            if (!string.IsNullOrWhiteSpace(authenticationID))
            {
                try
                {
                    MemoryStream buffer = new MemoryStream();
                    byte[] salt = new byte[DataPublisher.CipherSaltLength];
                    byte[] bytes;

                    // Generate some random prefix data to make sure auth key transmission is always unique
                    GSF.Security.Cryptography.Random.GetBytes(salt);

                    // Get encoded bytes of authentication key
                    bytes = salt.Combine(m_encoding.GetBytes(authenticationID));

                    // Encrypt authentication key
                    bytes = bytes.Encrypt(sharedSecret, CipherStrength.Aes256);

                    // Write encoded authentication key length into buffer
                    buffer.Write(EndianOrder.BigEndian.GetBytes(bytes.Length), 0, 4);

                    // Encode encrypted authentication key into buffer
                    buffer.Write(bytes, 0, bytes.Length);

                    // Send authentication command to server with associated command buffer
                    return SendServerCommand(ServerCommand.Authenticate, buffer.ToArray());
                }
                catch (Exception ex)
                {
                    LogDispatcher.E(StrLogSource, "ProcessException", "Exception occurred while trying to authenticate publisher subscription: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
                }
            }
            else
                LogDispatcher.E(StrLogSource, "ProcessException", "Cannot authenticate subscription without a connection string.");

            return false;
        }

        /// <summary>
        /// Subscribes (or re-subscribes) to a data publisher for a set of data points.
        /// </summary>
        /// <param name="info">Configuration object that defines the subscription.</param>
        /// <returns><c>true</c> if subscribe transmission was successful; otherwise <c>false</c>.</returns>
        public bool Subscribe(SubscriptionInfo info)
        {
            if (info is UnsynchronizedSubscriptionInfo)
                return UnsynchronizedSubscribe((UnsynchronizedSubscriptionInfo)info);

            throw new NotSupportedException("Type of subscription used is not supported");
        }

        /// <summary>
        /// Subscribes (or re-subscribes) to a data publisher for an unsynchronized set of data points.
        /// </summary>
        /// <param name="info">Configuration object that defines the subscription.</param>
        /// <returns><c>true</c> if subscribe transmission was successful; otherwise <c>false</c>.</returns>
        public bool UnsynchronizedSubscribe(UnsynchronizedSubscriptionInfo info)
        {
            StringBuilder connectionString = new StringBuilder();
            AssemblyInfo assemblyInfo = AssemblyInfo.ExecutingAssembly;

            connectionString.AppendFormat("trackLatestMeasurements={0};", info.Throttled);
            connectionString.AppendFormat("publishInterval={0};", info.PublishInterval);
            connectionString.AppendFormat("includeTime={0};", info.IncludeTime);
            connectionString.AppendFormat("lagTime={0};", info.LagTime);
            connectionString.AppendFormat("leadTime={0};", info.LeadTime);
            connectionString.AppendFormat("useLocalClockAsRealTime={0};", info.UseLocalClockAsRealTime);
            connectionString.AppendFormat("processingInterval={0};", info.ProcessingInterval);
            connectionString.AppendFormat("useMillisecondResolution={0};", info.UseMillisecondResolution);
            connectionString.AppendFormat("assemblyInfo={{source={0};version={1}.{2}.{3};buildDate={4}}};", assemblyInfo.Name, assemblyInfo.Version.Major, assemblyInfo.Version.Minor, assemblyInfo.Version.Build, assemblyInfo.BuildDate.ToString("yyyy-MM-dd HH:mm:ss"));

            if (!string.IsNullOrWhiteSpace(info.FilterExpression))
                connectionString.AppendFormat("inputMeasurementKeys={{{0}}};", info.FilterExpression);

            if (info.UdpDataChannel)
                connectionString.AppendFormat("dataChannel={{localport={0}}};", info.DataChannelLocalPort);

            if (!string.IsNullOrWhiteSpace(info.StartTime))
                connectionString.AppendFormat("startTimeConstraint={0};", info.StartTime);

            if (!string.IsNullOrWhiteSpace(info.StopTime))
                connectionString.AppendFormat("stopTimeConstraint={0};", info.StopTime);

            if (!string.IsNullOrWhiteSpace(info.ConstraintParameters))
                connectionString.AppendFormat("timeConstraintParameters={0};", info.ConstraintParameters);

            if (!string.IsNullOrWhiteSpace(info.WaitHandleNames))
            {
                connectionString.AppendFormat("waitHandleNames={0};", info.WaitHandleNames);
                connectionString.AppendFormat("waitHandleTimeout={0};", info.WaitHandleTimeout);
            }

            if (!string.IsNullOrWhiteSpace(info.ExtraConnectionStringParameters))
                connectionString.AppendFormat("{0};", info.ExtraConnectionStringParameters);

            // Make sure not to monitor for data loss any faster than downsample time on throttled connections - additionally
            // you will want to make sure data stream monitor is twice lagtime to allow time for initial points to arrive.
            if (info.Throttled && (object)m_dataStreamMonitor != null && m_dataStreamMonitor.Interval / 1000.0D < info.LagTime)
                m_dataStreamMonitor.Interval = 2.0D * info.LagTime * 1000.0D;

            // Set millisecond resolution member variable for compact measurement parsing
            Info.UseMillisecondResolution = info.UseMillisecondResolution;

            return Subscribe(false, info.UseCompactMeasurementFormat, connectionString.ToString());
        }

        /// <summary>
        /// Subscribes (or re-subscribes) to a data publisher for a set of data points.
        /// </summary>
        /// <param name="remotelySynchronized">Boolean value that determines if subscription should be remotely synchronized - note that data publisher may not allow remote synchronization.</param>
        /// <param name="compactFormat">Boolean value that determines if the compact measurement format should be used. Set to <c>false</c> for full fidelity measurement serialization; otherwise set to <c>true</c> for bandwidth conservation.</param>
        /// <param name="connectionString">Connection string that defines required and optional parameters for the subscription.</param>
        /// <returns><c>true</c> if subscribe transmission was successful; otherwise <c>false</c>.</returns>
        public virtual bool Subscribe(bool remotelySynchronized, bool compactFormat, string connectionString)
        {
            bool success = false;

            if (!string.IsNullOrWhiteSpace(connectionString))
            {
                try
                {
                    // Parse connection string to see if it contains a data channel definition
                    Dictionary<string, string> settings = connectionString.ParseKeyValuePairs();
                    UdpClient dataChannel = null;
                    string setting;

                    settings.TryGetValue("dataChannel", out setting);

                    if (!string.IsNullOrWhiteSpace(setting))
                    {
                        dataChannel = new UdpClient(setting);

                        dataChannel.ReceiveBufferSize = ushort.MaxValue;
                        dataChannel.MaxConnectionAttempts = -1;
                        dataChannel.ConnectAsync();
                    }

                    // Assign data channel client reference and attach to needed events
                    this.DataChannel = dataChannel;

                    // Setup subcription packet
                    MemoryStream buffer = new MemoryStream();
                    DataPacketFlags flags = DataPacketFlags.NoFlags;
                    byte[] bytes;

                    if (remotelySynchronized)
                        flags |= DataPacketFlags.Synchronized;

                    if (compactFormat)
                        flags |= DataPacketFlags.Compact;

                    // Write data packet flags into buffer
                    buffer.WriteByte((byte)flags);

                    // Get encoded bytes of connection string
                    bytes = m_encoding.GetBytes(connectionString);

                    // Write encoded connection string length into buffer
                    buffer.Write(EndianOrder.BigEndian.GetBytes(bytes.Length), 0, 4);

                    // Encode connection string into buffer
                    buffer.Write(bytes, 0, bytes.Length);

                    // Cache subscribed synchronization state
                    m_synchronizedSubscription = remotelySynchronized;

                    // Send subscribe server command with associated command buffer
                    success = SendServerCommand(ServerCommand.Subscribe, buffer.ToArray());
                }
                catch (Exception ex)
                {
                    LogDispatcher.E(StrLogSource, "ProcessException", "Exception occurred while trying to make publisher subscription: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
                }
            }
            else
                LogDispatcher.E(StrLogSource, "ProcessException", "Cannot make publisher subscription without a connection string.");

            return success;
        }

        /// <summary>
        /// Unsubscribes from a data publisher.
        /// </summary>
        /// <returns><c>true</c> if unsubscribe transmission was successful; otherwise <c>false</c>.</returns>
        public virtual bool Unsubscribe()
        {
            // Send unsubscribe server command
            return SendServerCommand(ServerCommand.Unsubscribe, null);
        }

        /// <summary>
        /// Returns the measurements signal IDs that were authorized after the last successful subscription request.
        /// </summary>
        [AdapterCommand("Gets authorized signal IDs from last subscription request.")]
        public virtual Guid[] GetAuthorizedSignalIDs()
        {
            if (m_signalIndexCache != null)
                return m_signalIndexCache.AuthorizedSignalIDs;

            return new Guid[0];
        }

        /// <summary>
        /// Returns the measurements signal IDs that were unauthorized after the last successful subscription request.
        /// </summary>
        [AdapterCommand("Gets unauthorized signal IDs from last subscription request.")]
        public virtual Guid[] GetUnauthorizedSignalIDs()
        {
            if (m_signalIndexCache != null)
                return m_signalIndexCache.UnauthorizedSignalIDs;

            return new Guid[0];
        }

        /// <summary>
        /// Initiate a metadata refresh.
        /// </summary>
        [AdapterCommand("Initiates a metadata refresh.")]
        public virtual void RefreshMetadata()
        {
            SendServerCommand(ServerCommand.MetaDataRefresh);
        }

        /// <summary>
        /// Sends a server command to the publisher connection.
        /// </summary>
        /// <param name="commandCode"><see cref="ServerCommand"/> to send.</param>
        /// <param name="data">Command data to send.</param>
        /// <returns><c>true</c> if <paramref name="commandCode"/> transmission was successful; otherwise <c>false</c>.</returns>
        public virtual bool SendServerCommand(ServerCommand commandCode, byte[] data = null)
        {
            if (m_commandChannel != null && m_commandChannel.CurrentState == ClientState.Connected)
            {
                try
                {
                    MemoryStream commandPacket = new MemoryStream();

                    // Write command code into command packet
                    commandPacket.WriteByte((byte)commandCode);

                    // Write command buffer into command packet
                    if (data != null && data.Length > 0)
                        commandPacket.Write(data, 0, data.Length);

                    // Send command packet to publisher
                    m_commandChannel.SendAsync(commandPacket.ToArray(), 0, (int)commandPacket.Length);

                    // Track server command in pending request queue
                    lock (m_requests)
                    {
                        // Make sure a pending request does not already exist
                        int index = m_requests.BinarySearch(commandCode);

                        if (index < 0)
                        {
                            // Add the new server command to the request list
                            m_requests.Add(commandCode);

                            // Make sure requests are sorted to allow for binary searching
                            m_requests.Sort();
                        }
                    }

                    return true;
                }
                catch (Exception ex)
                {
                    LogDispatcher.E(StrLogSource, "ProcessException", "Exception occurred while trying to send server command \"{0}\" to publisher: {1}{2}{3}", commandCode, ex.Message, Environment.NewLine, ex.StackTrace);
                }
            }
            else
                LogDispatcher.E(StrLogSource, "ProcessException", "Subscriber is currently unconnected. Cannot send server command \"{0}\" to publisher.", commandCode);

            return false;
        }

        public override void Start()
        {
            m_commandChannelConnectionAttempts = 0;
            m_dataChannelConnectionAttempts = 0;
            m_commandChannel.ConnectAsync();
            m_subscribed = false;
            m_keyIVs = null;
            m_totalBytesReceived = 0L;
            m_monitoredBytesReceived = 0L;
            m_lastBytesReceived = 0;
        }

        public override void Stop()
        {
            m_enabled = false;

            // Stop data stream monitor
            if ((object)m_dataStreamMonitor != null)
                m_dataStreamMonitor.Enabled = false;

            // Disconnect data channel
            if ((object)m_dataChannel != null)
                m_dataChannel.Disconnect();

            // Disconnect command channel
            if ((object)m_commandChannel != null)
                m_commandChannel.Disconnect();
        }

        /// <summary>
        /// Gets a short one-line status of this <see cref="DataSubscriber"/>.
        /// </summary>
        /// <param name="maxLength">Maximum length of the status message.</param>
        /// <returns>Text of the status message.</returns>
        public string GetShortStatus(int maxLength)
        {
            if (m_commandChannel != null && m_commandChannel.CurrentState == ClientState.Connected)
                return string.Format("Subscriber is connected and receiving {0} data points", m_synchronizedSubscription ? "synchronized" : "unsynchronized").CenterText(maxLength);

            return "Subscriber is not connected.".CenterText(maxLength);
        }

        /// <summary>
        /// Get message from string based response.
        /// </summary>
        /// <param name="buffer">Response buffer.</param>
        /// <param name="startIndex">Start index of response message.</param>
        /// <param name="length">Length of response message.</param>
        /// <returns>Decoded response string.</returns>
        protected string InterpretResponseMessage(byte[] buffer, int startIndex, int length)
        {
            return m_encoding.GetString(buffer, startIndex, length);
        }

        // Restarts the subscriber.
        private void Restart()
        {
            try
            {
                // Disconnect data channel
                if ((object)m_dataChannel != null)
                    m_dataChannel.Disconnect();

                // Disconnect command channel
                if ((object)m_commandChannel != null)
                {
                    m_commandChannel.Disconnect();

                    // Reset state and connect command channel
                    m_commandChannelConnectionAttempts = 0;
                    m_dataChannelConnectionAttempts = 0;
                    m_commandChannel.ConnectAsync();
                    m_subscribed = false;
                    m_keyIVs = null;
                    m_totalBytesReceived = 0L;
                    m_monitoredBytesReceived = 0L;
                    m_lastBytesReceived = 0;
                }
            }
            catch (Exception ex)
            {
                LogDispatcher.E(StrLogSource, "ProcessException", ex);
            }
        }

        private void ProcessServerResponse(byte[] buffer, int length)
        {
            // Currently this work is done on the async socket completion thread, make sure work to be done is timely and if the response processing
            // is coming in via the command channel and needs to send a command back to the server, it should be done on a separate thread...
            if (buffer != null && length > 0)
            {
                try
                {
                    ServerResponse responseCode = (ServerResponse)buffer[0];
                    ServerCommand commandCode = (ServerCommand)buffer[1];
                    int responseLength = EndianOrder.BigEndian.ToInt32(buffer, 2);
                    int responseIndex = 6;
                    bool solicited = false;
                    byte[][][] keyIVs;

                    // See if this was a solicited response to a requested server command
                    if (responseCode.IsSolicited())
                    {
                        lock (m_requests)
                        {
                            int index = m_requests.BinarySearch(commandCode);

                            if (index >= 0)
                            {
                                solicited = true;
                                m_requests.RemoveAt(index);
                            }
                        }

                        // Disconnect any established UDP data channel upon successful unsubscribe
                        if (solicited && commandCode == ServerCommand.Unsubscribe && responseCode == ServerResponse.Succeeded)
                            DataChannel = null;
                    }

                    switch (responseCode)
                    {
                        case ServerResponse.Succeeded:
                            if (solicited)
                            {
                                switch (commandCode)
                                {
                                    case ServerCommand.Authenticate:
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Success code received in response to server command \"{0}\": {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                                        OnConnectionAuthenticated();
                                        break;
                                    case ServerCommand.Subscribe:
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Success code received in response to server command \"{0}\": {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                                        m_subscribed = true;
                                        if ((object)m_dataStreamMonitor != null)
                                            m_dataStreamMonitor.Enabled = true;
                                        break;
                                    case ServerCommand.Unsubscribe:
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Success code received in response to server command \"{0}\": {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                                        m_subscribed = false;
                                        if ((object)m_dataStreamMonitor != null)
                                            m_dataStreamMonitor.Enabled = false;
                                        break;
                                    case ServerCommand.RotateCipherKeys:
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Success code received in response to server command \"{0}\": {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                                        break;
                                    case ServerCommand.MetaDataRefresh:
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Success code received in response to server command \"{0}\": latest meta-data received.", commandCode);
                                        OnMetaDataReceived(DeserializeMetadata(buffer.BlockCopy(responseIndex, responseLength)));
                                        break;
                                }
                            }
                            else
                            {
                                switch (commandCode)
                                {
                                    case ServerCommand.MetaDataRefresh:
                                        // Meta-data refresh may be unsolicited
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Received server confirmation for unsolicited request to \"{0}\" command: latest meta-data received.", commandCode);
                                        OnMetaDataReceived(DeserializeMetadata(buffer.BlockCopy(responseIndex, responseLength)));
                                        break;
                                    case ServerCommand.RotateCipherKeys:
                                        // Key rotation may be unsolicited
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Received server confirmation for unsolicited request to \"{0}\" command: {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                                        break;
                                    case ServerCommand.Subscribe:
                                        LogDispatcher.I(StrLogSource, "StatusMessage", "Received unsolicited response to \"{0}\" command: {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                                        break;
                                    default:
                                        LogDispatcher.E(StrLogSource, "ProcessException", "Publisher sent a success code for an unsolicited server command: ", commandCode);
                                        break;
                                }
                            }
                            break;
                        case ServerResponse.Failed:
                            if (solicited)
                                LogDispatcher.I(StrLogSource, "StatusMessage", "Failure code received in response to server command \"{0}\": {1}", commandCode, InterpretResponseMessage(buffer, responseIndex, responseLength));
                            else
                                LogDispatcher.E(StrLogSource, "ProcessException", "Publisher sent a failed code for an unsolicited server command: {0}", commandCode);
                            break;
                        case ServerResponse.DataPacket:
                            long now = DateTime.UtcNow.Ticks;

                            // Deserialize data packet
                            List<CompactMeasurement> measurements = new List<CompactMeasurement>();
                            DataPacketFlags flags;
                            Ticks epoch;
                            int count;

                            Guid streamID;
                            Ticks timestamp;
                            MeasurementStateFlags measurementFlags;
                            IMeasurementValue value;

                            // Track total data packet bytes received from any channel
                            m_totalBytesReceived += m_lastBytesReceived;
                            m_monitoredBytesReceived += m_lastBytesReceived;

                            // Get data packet flags
                            flags = (DataPacketFlags)buffer[responseIndex];
                            responseIndex++;

                            int cipherIndex = (flags & DataPacketFlags.CipherIndex) > 0 ? 1 : 0;

                            // Decrypt data packet payload if keys are available
                            if (m_keyIVs != null)
                            {
                                // Get a local copy of volatile keyIVs reference since this can change at any time
                                keyIVs = m_keyIVs;

                                // Decrypt payload portion of data packet
                                buffer = Common.SymmetricAlgorithm.Decrypt(buffer, responseIndex, responseLength - 1, keyIVs[cipherIndex][0], keyIVs[cipherIndex][1]);
                                responseIndex = 0;
                            }

                            // Deserialize epoch used to compress timestamps in packet
                            epoch = EndianOrder.BigEndian.ToInt64(buffer, responseIndex);
                            responseIndex += 8;

                            // Deserialize number of measurements that follow
                            count = EndianOrder.BigEndian.ToInt32(buffer, responseIndex);
                            responseIndex += 4;

                            // Deserialize measurements
                            for (int i = 0; i < count; i++)
                            {
                                if ((object)m_signalIndexCache != null)
                                {
                                    // Deserialize compact measurement format
                                    CompactMeasurement measurement = new CompactMeasurement();
                                    Tuple<Guid, string, uint> key;
                                    ushort runtimeID;

                                    measurement.StateFlags = (CompactMeasurementStateFlags)buffer[responseIndex];
                                    responseIndex++;

                                    runtimeID = EndianOrder.BigEndian.ToUInt16(buffer, responseIndex);
                                    responseIndex += 2;

                                    measurement.Value = EndianOrder.BigEndian.ToSingle(buffer, responseIndex);
                                    responseIndex += 4;

                                    if (!UseMillisecondResolution)
                                    {
                                        measurement.Timestamp = epoch + EndianOrder.BigEndian.ToUInt32(buffer, responseIndex);
                                        responseIndex += 4;
                                    }
                                    else
                                    {
                                        measurement.Timestamp = epoch + (EndianOrder.BigEndian.ToUInt16(buffer, responseIndex) * Ticks.PerMillisecond);
                                        responseIndex += 2;
                                    }

                                    if (m_signalIndexCache.Reference.TryGetValue(runtimeID, out key))
                                    {
                                        measurement.SignalID = key.Item1;
                                        measurements.Add(measurement);
                                    }
                                }
                                else if (m_lastMissingCacheWarning + MissingCacheWarningInterval < now)
                                {
                                    if (m_lastMissingCacheWarning != 0L)
                                    {
                                        // Warning message for missing signal index cache
                                        LogDispatcher.W(StrLogSource, "StatusMessage", "WARNING: Signal index cache has not arrived. No compact measurements can be parsed.");
                                    }

                                    m_lastMissingCacheWarning = now;
                                }
                            }

                            if ((object)MessageBusProxy != null)
                            {
                                streamID = Info.OutgoingStreamIDs.First();
                                timestamp = DateTime.UtcNow.Ticks;
                                measurementFlags = MeasurementStateFlags.Normal;
                                value = new DataPublisherValue(measurements.ToList(), null);
                                MessageBusProxy.Send(new Measurement(streamID, timestamp, measurementFlags, value));
                            }
                            else
                            {
                                OnNewMeasurements(measurements);
                            }

                            break;
                        case ServerResponse.BufferBlock:
                            // Buffer block received - wrap as a buffer block measurement and expose back to consumer
                            uint sequenceNumber = EndianOrder.BigEndian.ToUInt32(buffer, responseIndex);
                            int cacheIndex = (int)(sequenceNumber - m_expectedBufferBlockSequenceNumber);
                            BufferBlockMeasurement bufferBlockMeasurement;
                            Tuple<Guid, string, uint> measurementKey;
                            ushort signalIndex;

                            // Check if this buffer block has already been processed (e.g., mistaken retransmission due to timeout)
                            if (cacheIndex >= 0 && (cacheIndex >= m_bufferBlockCache.Count || (object)m_bufferBlockCache[cacheIndex] == null))
                            {
                                // Send confirmation that buffer block is received
                                SendServerCommand(ServerCommand.ConfirmBufferBlock, buffer.BlockCopy(responseIndex, 4));

                                // Get measurement key from signal index cache
                                signalIndex = EndianOrder.BigEndian.ToUInt16(buffer, responseIndex + 4);

                                if (!m_signalIndexCache.Reference.TryGetValue(signalIndex, out measurementKey))
                                    throw new InvalidOperationException("Failed to find associated signal identification for runtime ID " + signalIndex);

                                // Skip the sequence number and signal index when creating the buffer block measurement
                                bufferBlockMeasurement = new BufferBlockMeasurement()
                                {
                                    StreamID = measurementKey.Item1,
                                    Buffer = buffer.BlockCopy(responseIndex + 6, responseLength - 6),
                                    Length = responseLength - 6
                                };

                                // Determine if this is the next buffer block in the sequence
                                if (sequenceNumber == m_expectedBufferBlockSequenceNumber)
                                {
                                    List<BufferBlockMeasurement> bufferBlockMeasurements = new List<BufferBlockMeasurement>();
                                    int i;

                                    // Add the buffer block measurement to the list of measurements to be published
                                    bufferBlockMeasurements.Add(bufferBlockMeasurement);
                                    m_expectedBufferBlockSequenceNumber++;

                                    // Add cached buffer block measurements to the list of measurements to be published
                                    for (i = 1; i < m_bufferBlockCache.Count; i++)
                                    {
                                        if ((object)m_bufferBlockCache[i] == null)
                                            break;

                                        bufferBlockMeasurements.Add(m_bufferBlockCache[i]);
                                        m_expectedBufferBlockSequenceNumber++;
                                    }

                                    // Remove published measurements from the buffer block queue
                                    if (m_bufferBlockCache.Count > 0)
                                        m_bufferBlockCache.RemoveRange(0, i);

                                    // Publish measurements
                                    MessageBusProxy.Send(new RawData(Info.OutgoingStreamIDs.First(), new DataPublisherValue(null, bufferBlockMeasurements)));
                                }
                                else
                                {
                                    // Ensure that the list has at least as many
                                    // elements as it needs to cache this measurement
                                    for (int i = m_bufferBlockCache.Count; i <= cacheIndex; i++)
                                        m_bufferBlockCache.Add(null);

                                    // Insert this buffer block into the proper location in the list
                                    m_bufferBlockCache[cacheIndex] = bufferBlockMeasurement;
                                }
                            }
                            break;
                        case ServerResponse.DataStartTime:
                            // Raise data start time event
                            OnDataStartTime(EndianOrder.BigEndian.ToInt64(buffer, responseIndex));
                            break;
                        case ServerResponse.ProcessingComplete:
                            // Raise input processing completed event
                            OnProcessingComplete(InterpretResponseMessage(buffer, responseIndex, responseLength));
                            break;
                        case ServerResponse.UpdateSignalIndexCache:
                            // Deserialize new signal index cache
                            m_remoteSignalIndexCache = DeserializeSignalIndexCache(buffer.BlockCopy(responseIndex, responseLength));
                            m_signalIndexCache = new SignalIndexCache(Info.DataSource, m_remoteSignalIndexCache);
                            break;
                        case ServerResponse.UpdateCipherKeys:
                            // Get active cipher index
                            m_cipherIndex = buffer[responseIndex++];

                            // Extract remaining response
                            byte[] bytes = buffer.BlockCopy(responseIndex, responseLength - 1);

                            // Deserialize new cipher keys
                            keyIVs = new byte[2][][];
                            keyIVs[EvenKey] = new byte[2][];
                            keyIVs[OddKey] = new byte[2][];

                            int index = 0;
                            int bufferLen;

                            // Read even key size
                            bufferLen = EndianOrder.BigEndian.ToInt32(bytes, index);
                            index += 4;

                            // Read even key
                            keyIVs[EvenKey][KeyIndex] = new byte[bufferLen];
                            Buffer.BlockCopy(bytes, index, keyIVs[EvenKey][KeyIndex], 0, bufferLen);
                            index += bufferLen;

                            // Read even initialization vector size
                            bufferLen = EndianOrder.BigEndian.ToInt32(bytes, index);
                            index += 4;

                            // Read even initialization vector
                            keyIVs[EvenKey][IVIndex] = new byte[bufferLen];
                            Buffer.BlockCopy(bytes, index, keyIVs[EvenKey][IVIndex], 0, bufferLen);
                            index += bufferLen;

                            // Read odd key size
                            bufferLen = EndianOrder.BigEndian.ToInt32(bytes, index);
                            index += 4;

                            // Read odd key
                            keyIVs[OddKey][KeyIndex] = new byte[bufferLen];
                            Buffer.BlockCopy(bytes, index, keyIVs[OddKey][KeyIndex], 0, bufferLen);
                            index += bufferLen;

                            // Read odd initialization vector size
                            bufferLen = EndianOrder.BigEndian.ToInt32(bytes, index);
                            index += 4;

                            // Read odd initialization vector
                            keyIVs[OddKey][IVIndex] = new byte[bufferLen];
                            Buffer.BlockCopy(bytes, index, keyIVs[OddKey][IVIndex], 0, bufferLen);
                            index += bufferLen;

                            // Exchange keys
                            m_keyIVs = keyIVs;

                            LogDispatcher.I(StrLogSource, "StatusMessage", "Successfully established new cipher keys for data packet transmissions.");
                            break;
                    }
                }
                catch (Exception ex)
                {
                    LogDispatcher.E(StrLogSource, "ProcessException", "Failed to process publisher response packet due to exception: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
                }
            }
        }

        // Handles auto-connection subscription initialization
        private void StartSubscription()
        {
            UnsynchronizedSubscriptionInfo subscriptionInfo = new UnsynchronizedSubscriptionInfo(false);
            StringBuilder filterExpression = new StringBuilder();
            string localPortSetting;
            int localPort;

            // If TCP command channel is defined separately, then base connection string defines data channel
            if (Info.Settings.ContainsKey("commandChannel"))
            {
                subscriptionInfo.UdpDataChannel = true;

                if (Info.Settings.TryGetValue("port", out localPortSetting) || Info.Settings.TryGetValue("localport", out localPortSetting))
                {
                    if (int.TryParse(localPortSetting, out localPort))
                        subscriptionInfo.DataChannelLocalPort = localPort;
                }
            }

            if ((object)Info.SubscribedMeasurements != null)
            {
                foreach (Guid id in Info.SubscribedMeasurements)
                {
                    if (filterExpression.Length > 0)
                        filterExpression.Append(';');

                    // Subscribe by associated Guid...
                    filterExpression.Append(id.ToString());
                }

                subscriptionInfo.UseCompactMeasurementFormat = true;
                subscriptionInfo.FilterExpression = filterExpression.ToString();

                // Start unsynchronized subscription
                UnsynchronizedSubscribe(subscriptionInfo);
            }
            else
            {
                LogDispatcher.W(StrLogSource, "StatusMessage", "WARNING: No measurements are currently defined for subscription.");
            }

            // Initiate meta-data refresh
            if (Info.SynchronizeMetadata)
                SendServerCommand(ServerCommand.MetaDataRefresh);
        }

        /// <summary>
        /// Handles auto-connection metadata synchronization to local system. 
        /// </summary>
        /// <param name="state"><see cref="DataSet"/> metadata collection passed into state parameter.</param>
        /// <remarks>
        /// This function is normally called from thread pool since synchronization can take some time.
        /// </remarks>
        protected virtual void SynchronizeMetadata(object state)
        {
            try
            {
                DataSet metadata = state as DataSet;

                if (metadata != null)
                {
                    // Track total meta-data synchronization process time
                    Ticks startTime = DateTime.UtcNow.Ticks;

                    // Open the configuration database using settings found in the config file
                    AdoDataConnection database = new AdoDataConnection("systemSettings");
                    IDbConnection connection = database.Connection;
                    string guidPrefix = database.DatabaseType == DatabaseType.Access ? "{" : "'";
                    string guidSuffix = database.DatabaseType == DatabaseType.Access ? "}" : "'";

                    // Query the actual record ID based on the known run-time ID for this subscriber device
                    int parentID = Convert.ToInt32(connection.ExecuteScalar(string.Format("SELECT SourceID FROM Runtime WHERE ID = {0} AND SourceTable='Device'", Info.ID)));

                    // Validate that the subscriber device is marked as a concentrator (we are about to associate children devices with it)
                    if (!connection.ExecuteScalar(string.Format("SELECT IsConcentrator FROM Device WHERE ID = {0}", parentID)).ToString().ParseBoolean())
                        connection.ExecuteNonQuery(string.Format("UPDATE Device SET IsConcentrator = 1 WHERE ID = {0}", parentID));

                    // Determine the active node ID - we cache this since this value won't change for the lifetime of this class
                    if (m_nodeID == Guid.Empty)
                        m_nodeID = Guid.Parse(connection.ExecuteScalar(string.Format("SELECT NodeID FROM IaonInputAdapter WHERE ID = {0}", Info.ID)).ToString());

                    // Determine the protocol record auto-inc ID value for the gateway transport protocol (GEP) - this value is also cached since it shouldn't change for the lifetime of this class
                    if (m_gatewayProtocolID == 0)
                        m_gatewayProtocolID = int.Parse(connection.ExecuteScalar("SELECT ID FROM Protocol WHERE Acronym='GatewayTransport'").ToString());

                    // Prefix all children devices with the name of the parent since the same device names could appear in different connections (helps keep device names unique)
                    string sourcePrefix = Info.Name + "!";
                    Dictionary<string, int> deviceIDs = new Dictionary<string, int>(StringComparer.InvariantCultureIgnoreCase);
                    string selectSql, insertSql, updateSql, deleteSql, deviceAcronym, signalTypeAcronym;
                    int deviceID;

                    // Check to see if data for the "DeviceDetail" table was included in the meta-data
                    if (metadata.Tables.Contains("DeviceDetail"))
                    {
                        List<Guid> uniqueIDs = new List<Guid>();

                        if (database.IsSqlite)
                            connection.ExecuteNonQuery("begin");

                        foreach (DataRow row in metadata.Tables["DeviceDetail"].Rows)
                        {
                            Guid uniqueID = Guid.Parse(row.Field<object>("UniqueID").ToString()); // adoDatabase.Guid(row, "UniqueID"); // row.Field<Guid>("UniqueID");

                            // Track unique device guid's in this meta-data session, we'll need to remove any old associated devices that no longer exist
                            uniqueIDs.Add(uniqueID);

                            // We will synchronize metadata only if the source owns this device and it's not defined as a concentrator (these should normally be filtered by publisher - but we check just in case).
                            if (row.Field<object>("OriginalSource") == null && !row["IsConcentrator"].ToNonNullString("0").ParseBoolean())
                            {
                                // Define query to determine if this device is already defined (this should always be based on the unique device Guid)
                                selectSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Device WHERE UniqueID = {0}", "deviceGuid");

                                if (Convert.ToInt32(connection.ExecuteScalar(selectSql, database.Guid(uniqueID))) == 0)
                                {
                                    // Insert new device record
                                    insertSql = database.ParameterizedQueryString("INSERT INTO Device(NodeID, ParentID, Acronym, Name, ProtocolID, IsConcentrator, Enabled, OriginalSource) " +
                                        "VALUES ({0}, {1}, {2}, {3}, {4}, 0, 1, {5})", "nodeID", "parentID", "acronym", "name", "protocolID", "originalSource");

                                    connection.ExecuteNonQuery(insertSql, database.Guid(m_nodeID), parentID, sourcePrefix + row.Field<string>("Acronym"), row.Field<string>("Name"), m_gatewayProtocolID,
                                        Info.Internal == true ? (object)DBNull.Value : string.IsNullOrEmpty(row.Field<string>("ParentAcronym")) ? sourcePrefix + row.Field<string>("Acronym") : sourcePrefix + row.Field<string>("ParentAcronym"));

                                    // Guids are normally auto-generated during insert - after insertion update the guid so that it matches the source data. Most of the database
                                    // scripts have triggers that support properly assigning the Guid during an insert, but this code ensures the Guid will always get assigned.
                                    updateSql = database.ParameterizedQueryString("UPDATE Device SET UniqueID = {0} WHERE Acronym = {1}", "uniqueID", "acronym");
                                    connection.ExecuteNonQuery(updateSql, database.Guid(uniqueID), sourcePrefix + row.Field<string>("Acronym"));
                                }
                                else
                                {
                                    // Update existing device record
                                    if (Info.Internal)
                                    {
                                        // Gateway is assuming ownership of the device records when the "interal" flag is true - this means the device's measurements can be forwarded to another party.
                                        // From a device record perspective, ownership is inferred by setting 'OriginalSource' to null.
                                        updateSql = database.ParameterizedQueryString("UPDATE Device SET Acronym = {0}, Name = {1}, OriginalSource = {2}, ProtocolID = {3} WHERE UniqueID = {4}", "acronym", "name", "originalSource", "protocolID", "uniqueID");
                                        connection.ExecuteNonQuery(updateSql, sourcePrefix + row.Field<string>("Acronym"), row.Field<string>("Name"), (object)DBNull.Value, m_gatewayProtocolID, database.Guid(uniqueID));
                                    }
                                    else
                                    {
                                        // When gateway doesn't own device records (i.e., the "interal" flag is false), this means the device's measurements can only be consumed locally. From a device
                                        // record perspective this means the 'OriginalSource' field is set to the acronym of the PDC or PMU that generated the source measurments. This field allows a
                                        // mirrored source restriction to be implemented later to ensure all devices in an output protocol came from the same original source connection.
                                        updateSql = database.ParameterizedQueryString("UPDATE Device SET Acronym = {0}, Name = {1}, ProtocolID = {2} WHERE UniqueID = {3}", "acronym", "name", "protocolID", "uniqueID");
                                        connection.ExecuteNonQuery(updateSql, sourcePrefix + row.Field<string>("Acronym"), row.Field<string>("Name"), m_gatewayProtocolID, database.Guid(uniqueID));
                                    }
                                }
                            }

                            // Capture local device ID auto-inc value for measurement association
                            selectSql = database.ParameterizedQueryString("SELECT ID FROM Device WHERE UniqueID = {0}", "deviceGuid");
                            deviceIDs[row.Field<string>("Acronym")] = Convert.ToInt32(connection.ExecuteScalar(selectSql, database.Guid(uniqueID)));
                        }

                        if (database.IsSqlite)
                            connection.ExecuteNonQuery("end");

                        // Remove any device records associated with this subscriber that no longer exist in the meta-data
                        if (uniqueIDs.Count > 0)
                        {
                            deleteSql = string.Format("DELETE FROM Device WHERE ParentID = {0} AND UniqueID NOT IN ({1})", parentID, uniqueIDs.Select(uniqueID => guidPrefix + uniqueID.ToString().ToLower() + guidSuffix).ToDelimitedString(", "));
                            connection.ExecuteNonQuery(deleteSql);
                        }
                    }

                    // Check to see if data for the "MeasurementDetail" table was included in the meta-data
                    if (metadata.Tables.Contains("MeasurementDetail"))
                    {
                        List<Guid> signalIDs = new List<Guid>();

                        // Load signal type ID's from local database associated with their acronym for proper signal type translation
                        Dictionary<string, int> signalTypeIDs = new Dictionary<string, int>(StringComparer.InvariantCultureIgnoreCase);

                        foreach (DataRow row in connection.RetrieveData(database.AdapterType, "SELECT ID, Acronym FROM SignalType").Rows)
                        {
                            signalTypeAcronym = row.Field<string>("Acronym");

                            if (!string.IsNullOrWhiteSpace(signalTypeAcronym))
                                signalTypeIDs[signalTypeAcronym] = row.ConvertField<int>("ID");
                        }

                        if (database.IsSqlite)
                            connection.ExecuteNonQuery("begin");

                        foreach (DataRow row in metadata.Tables["MeasurementDetail"].Rows)
                        {
                            // Verify that this measurement record is internal to the publisher (these should be filtered by the publisher, but we still check since we won't be able to subscribe to any measurements it doesn't own)
                            if (row["Internal"].ToNonNullString("1").ParseBoolean())
                            {
                                // Get device and signal type acronyms
                                deviceAcronym = row.Field<string>("DeviceAcronym") ?? string.Empty;
                                signalTypeAcronym = row.Field<string>("SignalAcronym") ?? string.Empty;

                                // Make sure we have an associated device and signal type already defined for the measurement
                                if (!string.IsNullOrWhiteSpace(deviceAcronym) && deviceIDs.ContainsKey(deviceAcronym) && !string.IsNullOrWhiteSpace(signalTypeAcronym) && signalTypeIDs.ContainsKey(signalTypeAcronym))
                                {
                                    // Prefix the tag name with the "updated" device name
                                    deviceID = deviceIDs[deviceAcronym];
                                    string pointTag = sourcePrefix + row.Field<string>("PointTag") ?? string.Empty;
                                    Guid signalID = Guid.Parse(row.Field<object>("SignalID").ToString()); // adoDatabase.Guid(row, "SignalID");  // row.Field<Guid>("SignalID");

                                    // Track unique measurement signal guid's in this meta-data session, we'll need to remove any old associated measurements that no longer exist
                                    signalIDs.Add(signalID);

                                    // Define query to determine if this measurement is already defined (this should always be based on the unique signal ID Guid)
                                    selectSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Measurement WHERE SignalID = {0}", "signalID");

                                    if (Convert.ToInt32(connection.ExecuteScalar(selectSql, database.Guid(signalID))) == 0)
                                    {
                                        // Insert new measurement record
                                        insertSql = database.ParameterizedQueryString("INSERT INTO Measurement (DeviceID, PointTag, SignalTypeID, SignalReference, Description, Internal, Subscribed, Enabled) VALUES ({0}, {1}, {2}, {3}, {4}, {5}, 0, 1)", "deviceID", "pointTag", "signalTypeID", "signalReference", "description", "internal");
                                        connection.ExecuteNonQuery(insertSql, 30, deviceID, pointTag, signalTypeIDs[signalTypeAcronym], sourcePrefix + row.Field<string>("SignalReference"), row.Field<string>("Description") ?? string.Empty, database.Bool(Info.Internal));

                                        // Guids are normally auto-generated during insert - after insertion update the guid so that it matches the source data. Most of the database
                                        // scripts have triggers that support properly assigning the Guid during an insert, but this code ensures the Guid will always get assigned.
                                        updateSql = database.ParameterizedQueryString("UPDATE Measurement SET SignalID = {0} WHERE PointTag = {1}", "signalID", "pointTag");
                                        connection.ExecuteNonQuery(updateSql, database.Guid(signalID), pointTag);
                                    }
                                    else
                                    {
                                        // Update existing measurement record. Note that this update assumes that measurements will remain associated with a static source device.
                                        updateSql = database.ParameterizedQueryString("UPDATE Measurement SET PointTag = {0}, SignalTypeID = {1}, SignalReference = {2}, Description = {3}, Internal = {4} WHERE SignalID = {5}", "pointTag", "signalTypeID", "signalReference", "description", "internal", "signalID");
                                        connection.ExecuteNonQuery(updateSql, pointTag, signalTypeIDs[signalTypeAcronym], sourcePrefix + row.Field<string>("SignalReference"), row.Field<string>("Description") ?? string.Empty, database.Bool(Info.Internal), database.Guid(signalID));
                                    }
                                }
                            }
                        }

                        if (database.IsSqlite)
                            connection.ExecuteNonQuery("end");

                        // Remove any measurement records associated with existing devices in this session but no longer exist in the meta-data
                        if (deviceIDs.Count > 0 && signalIDs.Count > 0)
                        {
                            deleteSql = string.Format("DELETE FROM Measurement WHERE DeviceID IN ({0}) AND SignalID NOT IN ({1})", deviceIDs.Values.ToDelimitedString(", "), signalIDs.Select(uniqueID => guidPrefix + uniqueID.ToString() + guidSuffix).ToDelimitedString(", "));
                            connection.ExecuteNonQuery(deleteSql);
                        }
                    }

                    // Check to see if data for the "PhasorDetail" table was included in the meta-data
                    if (metadata.Tables.Contains("PhasorDetail"))
                    {
                        Dictionary<int, int> maxSourceIndicies = new Dictionary<int, int>();
                        int sourceIndex;

                        // Phasor data is normally only needed so that the user can property generate a mirrored IEEE C37.118 output stream from the source data.
                        // This is necessary since, in this protocol, the phasors are described (i.e., labeled) as a unit (i.e., as a complex number) instead of
                        // as two distinct angle and magnitude measurements.

                        foreach (DataRow row in metadata.Tables["PhasorDetail"].Rows)
                        {
                            // Get device acronym
                            deviceAcronym = row.Field<string>("DeviceAcronym") ?? string.Empty;

                            // Make sure we have an associated device already defined for the phasor record
                            if (!string.IsNullOrWhiteSpace(deviceAcronym) && deviceIDs.ContainsKey(deviceAcronym))
                            {
                                deviceID = deviceIDs[deviceAcronym];

                                // Define query to determine if this phasor record is already defined, this is no Guid for these simple label records
                                selectSql = database.ParameterizedQueryString("SELECT COUNT(*) FROM Phasor WHERE DeviceID = {0} AND SourceIndex = {1}", "deviceID", "sourceIndex");

                                if (Convert.ToInt32(connection.ExecuteScalar(selectSql, 30, deviceID, row.ConvertField<int>("SourceIndex"))) == 0)
                                {
                                    // Insert new phasor record
                                    insertSql = database.ParameterizedQueryString("INSERT INTO Phasor (DeviceID, Label, Type, Phase, SourceIndex) VALUES ({0}, {1}, {2}, {3}, {4})", "deviceID", "label", "type", "phase", "sourceIndex");
                                    connection.ExecuteNonQuery(insertSql, 30, deviceID, row.Field<string>("Label") ?? "undefined", (row.Field<string>("Type") ?? "V").TruncateLeft(1), (row.Field<string>("Phase") ?? "+").TruncateLeft(1), row.ConvertField<int>("SourceIndex"));
                                }
                                else
                                {
                                    // Update existing phasor record
                                    updateSql = database.ParameterizedQueryString("UPDATE Phasor SET Label = {0}, Type = {1}, Phase = {2} WHERE DeviceID = {3} AND SourceIndex = {4}", "label", "type", "phase", "deviceID", "sourceIndex");
                                    connection.ExecuteNonQuery(updateSql, row.Field<string>("Label") ?? "undefined", (row.Field<string>("Type") ?? "V").TruncateLeft(1), (row.Field<string>("Phase") ?? "+").TruncateLeft(1), deviceID, row.ConvertField<int>("SourceIndex"));
                                }

                                // Track largest source index for each device
                                maxSourceIndicies.TryGetValue(deviceID, out sourceIndex);

                                if (row.ConvertField<int>("SourceIndex") > sourceIndex)
                                    maxSourceIndicies[deviceID] = row.ConvertField<int>("SourceIndex");
                            }
                        }

                        // Remove any phasor records associated with existing devices in this session but no longer exist in the meta-data
                        if (maxSourceIndicies.Count > 0)
                        {
                            foreach (KeyValuePair<int, int> deviceIndexPair in maxSourceIndicies)
                            {
                                deleteSql = string.Format("DELETE FROM Phasor WHERE DeviceID = {0} AND SourceIndex > {1}", deviceIndexPair.Key, deviceIndexPair.Value);
                                connection.ExecuteNonQuery(deleteSql);
                            }
                        }
                    }

                    // New signals may have been defined, take original remote signal index cache and apply changes
                    if (m_remoteSignalIndexCache != null)
                        m_signalIndexCache = new SignalIndexCache(Info.DataSource, m_remoteSignalIndexCache);

                    LogDispatcher.I(StrLogSource, "StatusMessage", "Meta-data synchronization completed successfully in {0}", (DateTime.UtcNow.Ticks - startTime).ToElapsedTimeString(3));
                }
                else
                {
                    LogDispatcher.W(StrLogSource, "StatusMessage", "WARNING: Meta-data synchronization was not performed, deserialized dataset was empty.");
                }
            }
            catch (Exception ex)
            {
                LogDispatcher.E(StrLogSource, "ProcessException", "Failed to synchronize meta-data to local cache: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        private SignalIndexCache DeserializeSignalIndexCache(byte[] buffer)
        {
            GatewayCompressionMode gatewayCompressionMode = (GatewayCompressionMode)(Info.OperationalModes & OperationalModes.CompressionModeMask);
            bool useCommonSerializationFormat = (Info.OperationalModes & OperationalModes.UseCommonSerializationFormat) > 0;
            bool compressSignalIndexCache = (Info.OperationalModes & OperationalModes.CompressSignalIndexCache) > 0;

            SignalIndexCache deserializedCache;

            MemoryStream compressedData = null;
            GZipStream inflater = null;

            if (compressSignalIndexCache && gatewayCompressionMode == GatewayCompressionMode.GZip)
            {
                try
                {
                    compressedData = new MemoryStream(buffer);
                    inflater = new GZipStream(compressedData, CompressionMode.Decompress);
                    buffer = inflater.ReadStream();
                }
                finally
                {
                    if ((object)inflater != null)
                        inflater.Close();

                    if ((object)compressedData != null)
                        compressedData.Close();
                }
            }

            if (useCommonSerializationFormat)
            {
                deserializedCache = new SignalIndexCache();
                deserializedCache.Encoding = m_encoding;
                deserializedCache.ParseBinaryImage(buffer, 0, buffer.Length);
            }
            else
            {
                deserializedCache = Serialization.Deserialize<SignalIndexCache>(buffer, GSF.SerializationFormat.Binary);
            }

            return deserializedCache;
        }

        private DataSet DeserializeMetadata(byte[] buffer)
        {
            GatewayCompressionMode gatewayCompressionMode = (GatewayCompressionMode)(Info.OperationalModes & OperationalModes.CompressionModeMask);
            bool useCommonSerializationFormat = (Info.OperationalModes & OperationalModes.UseCommonSerializationFormat) > 0;
            bool compressMetadata = (Info.OperationalModes & OperationalModes.CompressMetadata) > 0;

            DataSet deserializedMetadata;

            MemoryStream compressedData = null;
            GZipStream inflater = null;

            MemoryStream encodedData = null;
            XmlTextReader unicodeReader = null;

            if (compressMetadata && gatewayCompressionMode == GatewayCompressionMode.GZip)
            {
                try
                {
                    // Insert compressed data into compressed buffer
                    compressedData = new MemoryStream(buffer);
                    inflater = new GZipStream(compressedData, CompressionMode.Decompress);
                    buffer = inflater.ReadStream();
                }
                finally
                {
                    if ((object)inflater != null)
                        inflater.Close();

                    if ((object)compressedData != null)
                        compressedData.Close();
                }
            }

            if (useCommonSerializationFormat)
            {
                try
                {
                    // Copy decompressed data into encoded buffer
                    encodedData = new MemoryStream(buffer);

                    // Read encoded data into data set as XML
                    unicodeReader = new XmlTextReader(encodedData);
                    deserializedMetadata = new DataSet();
                    deserializedMetadata.ReadXml(unicodeReader, XmlReadMode.ReadSchema);
                }
                finally
                {
                    if ((object)unicodeReader != null)
                        unicodeReader.Close();

                    if ((object)encodedData != null)
                        encodedData.Close();
                }
            }
            else
            {
                deserializedMetadata = Serialization.Deserialize<DataSet>(buffer, GSF.SerializationFormat.Binary);
            }

            return deserializedMetadata;
        }

        private Encoding GetCharacterEncoding(OperationalEncoding operationalEncoding)
        {
            Encoding encoding;

            switch (operationalEncoding)
            {
                case OperationalEncoding.Unicode:
                    encoding = Encoding.Unicode;
                    break;
                case OperationalEncoding.BigEndianUnicode:
                    encoding = Encoding.BigEndianUnicode;
                    break;
                case OperationalEncoding.UTF8:
                    encoding = Encoding.UTF8;
                    break;
                case OperationalEncoding.ANSI:
                    encoding = Encoding.Default;
                    break;
                default:
                    throw new InvalidOperationException(string.Format("Unsupported encoding detected: {0}", operationalEncoding));
            }

            return encoding;
        }

        // Socket exception handler
        private bool HandleSocketException(Exception ex)
        {
            SocketException socketException = ex as SocketException;

            if ((object)socketException != null)
            {
                // WSAECONNABORTED and WSAECONNRESET are common errors after a client disconnect,
                // if they happen for other reasons, make sure disconnect procedure is handled
                if (socketException.ErrorCode == 10053 || socketException.ErrorCode == 10054)
                {
                    DisconnectClient();
                    return true;
                }
            }

            if ((object)ex != null)
                HandleSocketException(ex.InnerException);

            return false;
        }

        // Disconnect client, restarting if disconnect was not intentional
        private void DisconnectClient()
        {
            DataChannel = null;

            // If user didn't initiate disconnect, restart the connection
            if (m_enabled)
                Restart();
        }

        // This method is called when connection has been authenticated
        private void DataSubscriber_ConnectionAuthenticated(object sender, EventArgs e)
        {
            if (Info.AutoConnect)
                StartSubscription();
        }

        // This method is called then new metadata has been received
        private void DataSubscriber_MetaDataReceived(object sender, EventArgs<DataSet> e)
        {
            try
            {
                // We handle synchronization on a seperate thread since this process may be lengthy
                if (Info.SynchronizeMetadata)
                    ThreadPool.QueueUserWorkItem(SynchronizeMetadata, e.Argument);
            }
            catch (Exception ex)
            {
                // Process exception for logging
                LogDispatcher.E(StrLogSource, "ProcessException", "Failed to queue metadata synchronization due to exception: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        /// <summary>
        /// Raises the <see cref="ConnectionEstablished"/> event.
        /// </summary>
        protected void OnConnectionEstablished()
        {
            try
            {
                if (ConnectionEstablished != null)
                    ConnectionEstablished(this, EventArgs.Empty);
            }
            catch (Exception ex)
            {
                // We protect our code from consumer thrown exceptions
                LogDispatcher.E(StrLogSource, "ProcessException", "Exception in consumer handler for ConnectionEstablished event: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        /// <summary>
        /// Raises the <see cref="ConnectionTerminated"/> event.
        /// </summary>
        protected void OnConnectionTerminated()
        {
            try
            {
                if (ConnectionTerminated != null)
                    ConnectionTerminated(this, EventArgs.Empty);
            }
            catch (Exception ex)
            {
                // We protect our code from consumer thrown exceptions
                LogDispatcher.E(StrLogSource, "ProcessException", "Exception in consumer handler for ConnectionTerminated event: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        /// <summary>
        /// Raises the <see cref="ConnectionAuthenticated"/> event.
        /// </summary>
        protected void OnConnectionAuthenticated()
        {
            try
            {
                if (ConnectionAuthenticated != null)
                    ConnectionAuthenticated(this, EventArgs.Empty);
            }
            catch (Exception ex)
            {
                // We protect our code from consumer thrown exceptions
                LogDispatcher.E(StrLogSource, "ProcessException", "Exception in consumer handler for ConnectionAuthenticated event: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        /// <summary>
        /// Raises the <see cref="MetaDataReceived"/> event.
        /// </summary>
        /// <param name="metadata">Meta-data <see cref="DataSet"/> instance to send to client subscription.</param>
        protected void OnMetaDataReceived(DataSet metadata)
        {
            try
            {
                if (MetaDataReceived != null)
                    MetaDataReceived(this, new EventArgs<DataSet>(metadata));
            }
            catch (Exception ex)
            {
                // We protect our code from consumer thrown exceptions
                LogDispatcher.E(StrLogSource, "ProcessException", "Exception in consumer handler for MetaDataReceived event: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        /// <summary>
        /// Raises the <see cref="DataStartTime"/> event.
        /// </summary>
        /// <param name="startTime">Start time, in <see cref="Ticks"/>, of first measurement transmitted.</param>
        protected void OnDataStartTime(Ticks startTime)
        {
            try
            {
                if (DataStartTime != null)
                    DataStartTime(this, new EventArgs<Ticks>(startTime));
            }
            catch (Exception ex)
            {
                // We protect our code from consumer thrown exceptions
                LogDispatcher.E(StrLogSource, "ProcessException", "Exception in consumer handler for DataStartTime event: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        /// <summary>
        /// Raises the <see cref="ProcessingComplete"/> event.
        /// </summary>
        /// <param name="source">Type name of adapter that sent the processing completed notification.</param>
        protected void OnProcessingComplete(string source)
        {
            try
            {
                if (ProcessingComplete != null)
                    ProcessingComplete(this, new EventArgs<string>(source));
            }
            catch (Exception ex)
            {
                // We protect our code from consumer thrown exceptions
                LogDispatcher.E(StrLogSource, "ProcessException", "Exception in consumer handler for ProcessingComplete event: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        private void m_dataStreamMonitor_Elapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            if (m_monitoredBytesReceived == 0)
            {
                // If we've received no data in the last timespan, we restart connect cycle...
                m_dataStreamMonitor.Enabled = false;
                LogDispatcher.I(StrLogSource, "StatusMessage", "{0}No data received in {1} seconds, restarting connect cycle...{2}", Environment.NewLine, (m_dataStreamMonitor.Interval / 1000.0D).ToString("0.0"), Environment.NewLine);
                ThreadPool.QueueUserWorkItem(state => Restart());
            }

            // Reset bytes received bytes being monitored
            m_monitoredBytesReceived = 0L;
        }

        private void OnNewMeasurements(List<CompactMeasurement> measurements)
        {
            if ((object)NewMeasurements != null)
                NewMeasurements(this, new EventArgs<List<CompactMeasurement>>(measurements));
        }

        #region [ Command Channel Event Handlers ]

        private void m_commandChannel_ConnectionEstablished(object sender, EventArgs e)
        {
            // Make sure no existing requests are queued for a new publisher connection
            lock (m_requests)
            {
                m_requests.Clear();
            }

            m_enabled = true;

            // Define operational modes as soon as possible
            SendServerCommand(ServerCommand.DefineOperationalModes, EndianOrder.BigEndian.GetBytes((uint)Info.OperationalModes));

            // Notify consumer that connection was sucessfully established
            OnConnectionEstablished();

            LogDispatcher.I(StrLogSource, "StatusMessage", "Data subscriber command channel connection to publisher was established.");

            if (Info.AutoConnect)
            {
                // Attempt authentication if required, remaining steps will happen on successful authentication
                StartSubscription();
            }
        }

        private void m_commandChannel_ConnectionTerminated(object sender, EventArgs e)
        {
            OnConnectionTerminated();
            LogDispatcher.I(StrLogSource, "StatusMessage", "Data subscriber command channel connection to publisher was terminated.");
            DisconnectClient();
        }

        private void m_commandChannel_ConnectionException(object sender, EventArgs<Exception> e)
        {
            Exception ex = e.Argument;
            if (ex is SecurityException)
            {
                LogDispatcher.E(StrLogSource, "Security", ex.Message);
            }
            LogDispatcher.E(StrLogSource, "ProcessException", new InvalidOperationException(string.Format("Data subscriber encountered an exception while attempting command channel publisher connection: {0}", ex.Message), ex));
        }

        private void m_commandChannel_ConnectionAttempt(object sender, EventArgs e)
        {
            // Inject a short delay between multiple connection attempts
            if (m_commandChannelConnectionAttempts > 0)
                Thread.Sleep(2000);

            LogDispatcher.I(StrLogSource, "StatusMessage", "Attempting command channel connection to publisher...");
            m_commandChannelConnectionAttempts++;
        }

        private void m_commandChannel_SendDataException(object sender, EventArgs<Exception> e)
        {
            Exception ex = e.Argument;
            if (ex is SecurityException)
            {
                LogDispatcher.E(StrLogSource, "Security", ex.Message);
            }
            if (!HandleSocketException(ex) && !(ex is ObjectDisposedException))
                LogDispatcher.I(StrLogSource, "ProcessException", "Data subscriber encountered an exception while sending command channel data to publisher connection: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
        }

        private void m_commandChannel_ReceiveData(object sender, EventArgs<int> e)
        {
            try
            {
                byte[] buffer;
                int length = e.Argument;

                m_lastBytesReceived = length;

                buffer = BufferPool.TakeBuffer(length);

                try
                {
                    m_commandChannel.Read(buffer, 0, length);
                    ProcessServerResponse(buffer, length);
                }
                finally
                {
                    if (buffer != null)
                        BufferPool.ReturnBuffer(buffer);
                }
            }
            catch (Exception ex)
            {
                LogDispatcher.E(StrLogSource, "ProcessException", "{0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        private void m_commandChannel_ReceiveDataException(object sender, EventArgs<Exception> e)
        {
            Exception ex = e.Argument;
            if (ex is SecurityException)
            {
                LogDispatcher.E(StrLogSource, "Security", ex.Message);
            }
            if (!HandleSocketException(ex) && !(ex is ObjectDisposedException))
                LogDispatcher.E(StrLogSource,
                                "ProcessException",
                                "Data subscriber encountered an exception while receiving command channel data from publisher connection: {0}{1}{2}",
                                ex.Message, Environment.NewLine, ex.StackTrace);
        }

        #endregion

        #region [ Data Channel Event Handlers ]

        private void m_dataChannel_ConnectionException(object sender, EventArgs<Exception> e)
        {
            Exception ex = e.Argument;
            if (ex is SecurityException)
            {
                LogDispatcher.E(StrLogSource, "Security", ex.Message);
            }
            LogDispatcher.I(StrLogSource,
                            "ProcessException",
                            "Data subscriber encountered an exception while attempting to establish data channel connection: {0}{1}{2}",
                            ex.Message, Environment.NewLine, ex.StackTrace);
        }

        private void m_dataChannel_ConnectionAttempt(object sender, EventArgs e)
        {
            // Inject a short delay between multiple connection attempts
            if (m_dataChannelConnectionAttempts > 0)
                Thread.Sleep(2000);

            LogDispatcher.I(StrLogSource, "StatusMessage", "Attempting to establish data channel connection to publisher...");
            m_dataChannelConnectionAttempts++;
        }

        private void m_dataChannel_ReceiveData(object sender, EventArgs<int> e)
        {
            try
            {
                byte[] buffer;
                int length = e.Argument;

                m_lastBytesReceived = length;

                buffer = BufferPool.TakeBuffer(length);

                try
                {
                    m_dataChannel.Read(buffer, 0, length);
                    ProcessServerResponse(buffer, length);
                }
                finally
                {
                    if (buffer != null)
                        BufferPool.ReturnBuffer(buffer);
                }
            }
            catch (Exception ex)
            {
                LogDispatcher.E(StrLogSource, "ProcessException", "{0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
            }
        }

        private void m_dataChannel_ReceiveDataException(object sender, EventArgs<Exception> e)
        {
            Exception ex = e.Argument;
            if (ex is SecurityException)
            {
                LogDispatcher.E(StrLogSource, "Security", ex.Message);
            }
            if (!HandleSocketException(ex) && !(ex is ObjectDisposedException))
                LogDispatcher.E(StrLogSource, "ProcessException", "Data subscriber encountered an exception while receiving data from publisher connection: {0}{1}{2}", ex.Message, Environment.NewLine, ex.StackTrace);
        }

        #endregion

        #endregion

        #region [ Static ]

        // Static Methods

        /// <summary>
        /// Maps <see cref="CompactMeasurementStateFlags"/> to <see cref="MeasurementStateFlags"/>.
        /// </summary>
        /// <param name="stateFlags">Flags to map.</param>
        /// <returns><see cref="MeasurementStateFlags"/> mapped from <see cref="CompactMeasurementStateFlags"/>.</returns>
        public static MeasurementStateFlags MapToFullFlags(CompactMeasurementStateFlags stateFlags)
        {
            MeasurementStateFlags mappedStateFlags = MeasurementStateFlags.Normal;

            if ((stateFlags & CompactMeasurementStateFlags.DataRange) > 0)
                mappedStateFlags |= DataRangeMask;

            if ((stateFlags & CompactMeasurementStateFlags.DataQuality) > 0)
                mappedStateFlags |= DataQualityMask;

            if ((stateFlags & CompactMeasurementStateFlags.TimeQuality) > 0)
                mappedStateFlags |= TimeQualityMask;

            if ((stateFlags & CompactMeasurementStateFlags.SystemIssue) > 0)
                mappedStateFlags |= SystemIssueMask;

            if ((stateFlags & CompactMeasurementStateFlags.CalculatedValue) > 0)
                mappedStateFlags |= CalculatedValueMask;

            if ((stateFlags & CompactMeasurementStateFlags.DiscardedValue) > 0)
                mappedStateFlags |= DiscardedValueMask;

            return mappedStateFlags;
        }

        #endregion
    }
}
